package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo25PageRank {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo25PageRank")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    // 读取文件 加载数据 并进行简单的处理
    val pageRDD: RDD[(String, List[String])] = sc.textFile("Spark/data/pageRank.txt")
      .map(line => {
        val splits: Array[String] = line.split("->")
        val page: String = splits(0)
        val linkPages: List[String] = splits(1).split(",").toList
        (page, linkPages)
      })

    // 给每个页面赋上初始的PR值 1
    var pagePRRDD: RDD[(String, List[String], Double)] = pageRDD.map(t2 => (t2._1, t2._2, 1.0))

    // 引入阻尼系数
    val q: Double = 0.85
    val N: Long = pageRDD.count()

    var flag: Boolean = true
    // 循环迭代
    while (flag) {
      // 将PR值 平均分给每一个初链 并使用flatMap展开
      val avgPRLinkPageRDD: RDD[(String, Double)] = pagePRRDD.flatMap(t3 => {
        val avgPR: Double = t3._3 / t3._2.size.toDouble
        t3._2.map(linkPage => (linkPage, avgPR)) // 将该页面的PR值平均分给每一个初链
      })

      // 统计每个页面的新的PR值
      val newPageRDD: RDD[(String, Double)] = avgPRLinkPageRDD.reduceByKey(_ + _)


      // 将新的PR值赋给每个页面
      val newPagePRRDD: RDD[(String, List[String], Double)] = pageRDD
        .join(newPageRDD)
        .map {
          case (page: String, (linkPages: List[String], newPR: Double)) =>
            (page, linkPages, (1 - q) / N + q * newPR)
        }

      newPagePRRDD.foreach(println)

      /**
       * pagePRRDD
       * (B,List(C),1.3333333333332575)
       * (A,List(B, D),0.761904761904816)
       * (C,List(A, B),1.5238095238094047)
       * (D,List(B, C),0.3809523809525217)
       *
       * newPagePRRDD
       * (B,List(C),1.333333333333485)
       * (A,List(B, D),0.7619047619050434)
       * (C,List(A, B),1.523809523809632)
       * (D,List(B, C),0.3809523809518396)
       *
       */
      // 让循环停止 计算 当前的PR值 跟 上一次计算的PR 的一个平均差值
      // 当 平均差值 <= 0.0001

      val pagePRKVRDD: RDD[(String, Double)] = pagePRRDD.map(t3 => (t3._1, t3._3))
      val avgDiffPR: Double = newPagePRRDD
        .map(t3 => (t3._1, t3._3))
        .join(pagePRKVRDD)
        .map {
          case (page: String, (newPR: Double, oldPR: Double)) =>
            Math.abs(newPR - oldPR) // 避免负数的情况 计算差值
        }.sum() / newPagePRRDD.count()

      if (avgDiffPR <= 0.0001) {
        // 符合条件表示收敛 跳出循环
        flag = false
      }

      pagePRRDD = newPagePRRDD

    }
  }

}
